package com.bawei.persona.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bawei.persona.realtime.app.func.TableProcessHoKFunction;
import com.bawei.persona.realtime.bean.TableProcess;
import com.bawei.persona.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

public class BaseDBApp {

    public static void main(String[] args) throws Exception {

        // 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(1);

        // 从kafka获取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "test");
        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer<>("ods_base_db_m", new SimpleStringSchema(), properties));

        // 结构转换
        SingleOutputStreamOperator<JSONObject> map = stream.map(json -> JSON.parseObject(json));
        map.print("----------------");

        // 数据处理
        SingleOutputStreamOperator<JSONObject> filter = map.filter(
                jsonObj -> {
                    boolean flag = jsonObj.getString("table") != null
                            && jsonObj.getJSONObject("data") != null
                            && jsonObj.getString("data").length() > 3;
                    return flag;
                }
        );

        // 分流器分流
        // 输出到HBASE
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>(TableProcess.SINK_TYPE_HBASE) {};
        SingleOutputStreamOperator<JSONObject> process = filter.process(new TableProcessHoKFunction(outputTag));

        FlinkKafkaProducer<JSONObject> kafkaSinkBySchema = MyKafkaUtil.getKafkaSinkBySchema(new KafkaSerializationSchema<JSONObject>() {

            @Override
            public void open(SerializationSchema.InitializationContext context) throws Exception {
                System.out.println("虚拟化kafkatopic数据");
            }

            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                String sink_table = jsonObject.getString("sink_table");
                String data = jsonObject.getString("data");
                return new ProducerRecord<>(sink_table, data.toString().getBytes());
            }
        });

        // 回写到dwd
        process.addSink(kafkaSinkBySchema);
        env.execute();

    }
}
